package cn.me.alphamq.common.model.client.spi.listener;

import cn.me.alphamq.common.model.client.TxCallbackCheckEvent;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.PushTxMsgReq;
import cn.me.alphamq.common.entity.req.TxCallbackCheckReq;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.enumeration.LocalTxStateEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.store.TxListenerManager;
import cn.me.alphamq.common.transaction.TransactionListener;
import com.alibaba.fastjson2.JSON;

/**
 * 事务回调检查的事件的监听器
 *
 * @author f
 */
public class TxCallbackCheckListener implements Listener<TxCallbackCheckEvent> {

    @Override
    public void onReceive(TxCallbackCheckEvent event) throws Exception {
        TxCallbackCheckReq txCallbackCheckReq = event.getTxCallbackCheckReq();
        String msgId = txCallbackCheckReq.getMsgId();
        String producerId = txCallbackCheckReq.getProducerId();
        TransactionListener txListener = TxListenerManager.get(producerId);
        Message message = Message.newCheckTxStateMsg(txCallbackCheckReq);
        LocalTxStateEnum localTxState = txListener.checkLocalTransaction(message);

        PushTxMsgReq pushTxMsgReq = PushTxMsgReq.newRemainMsg(message, msgId, producerId, localTxState);
        if (LocalTxStateEnum.COMMIT.equals(localTxState)) {
            event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_TX_MSG_REQ, JSON.toJSONBytes(pushTxMsgReq)));
        } else if (LocalTxStateEnum.ROLLBACK.equals(localTxState)) {
            event.getCtx().writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_TX_MSG_REQ, JSON.toJSONBytes(pushTxMsgReq)));
        } else if (LocalTxStateEnum.UNKNOWN.equals(localTxState)) {
            // 如果本地事务的状态仍是未知，则不做任何操作
        }
    }
}
